package com.marveal.rabbit.task.parser;

import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.executor.handler.JobProperties;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.marveal.rabbit.task.annotation.ElasticJobConfig;
import com.marveal.rabbit.task.autoconfigure.JobZookeeperProperties;
import com.marveal.rabbit.task.enums.ElasticJobTypeEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Map;

@Slf4j
public class ElasticJobConfigParser implements ApplicationListener<ApplicationReadyEvent> {

    private final JobZookeeperProperties jobZookeeperProperties;

    private final ZookeeperRegistryCenter zookeeperRegistryCenter;

    public ElasticJobConfigParser(
            JobZookeeperProperties jobZookeeperProperties,
            ZookeeperRegistryCenter zookeeperRegistryCenter) {
        this.jobZookeeperProperties = jobZookeeperProperties;
        this.zookeeperRegistryCenter = zookeeperRegistryCenter;
    }


    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        ConfigurableApplicationContext appCtx = event.getApplicationContext();
        Map<String, Object> beanMap = appCtx.getBeansWithAnnotation(ElasticJobConfig.class);
        for (Map.Entry<String, Object> entry : beanMap.entrySet()) {
            Object bean = entry.getValue();
            Class<?> clazz = bean.getClass();
            String className = clazz.getName();

            if (className.contains("$")) {
                int index = className.indexOf("$");
                clazz = getClassByName(className.substring(0, index));
            }
            String jobTypeName = clazz.getInterfaces()[0].getSimpleName();

            ElasticJobConfig elasticJobConfig = clazz.getAnnotation(ElasticJobConfig.class);
            className = clazz.getName();
            String jobName = jobZookeeperProperties.getNamespace() + "." + elasticJobConfig.jobName();
            String cron = elasticJobConfig.cron();
            String shardingItemParameters = elasticJobConfig.shardingItemParameters();
            String description = elasticJobConfig.description();
            String jobParameter = elasticJobConfig.jobParameter();
            String jobExceptionHandler = elasticJobConfig.jobExceptionHandler();
            String executorServiceHandler = elasticJobConfig.executorServiceHandler();

            String jobShardingStrategyClass = elasticJobConfig.jobShardingStrategyClass();
            String eventTraceRdbDataSource = elasticJobConfig.eventTraceRdbDataSource();
            String scriptCommandLine = elasticJobConfig.scriptCommandLine();

            boolean failover = elasticJobConfig.failover();
            boolean misfire = elasticJobConfig.misfire();
            boolean overwrite = elasticJobConfig.overwrite();
            boolean disabled = elasticJobConfig.disabled();
            boolean monitorExecution = elasticJobConfig.monitorExecution();
            boolean streamingProcess = elasticJobConfig.streamingProcess();

            int shardingTotalCount = elasticJobConfig.shardingTotalCount();
            int monitorPort = elasticJobConfig.monitorPort();
            int maxTimeDiffSeconds = elasticJobConfig.maxTimeDiffSeconds();
            int reconcileIntervalMinutes = elasticJobConfig.reconcileIntervalMinutes();

            // 创建JobCoreConfiguration
            JobCoreConfiguration coreConfiguration = JobCoreConfiguration
                    .newBuilder(jobName, cron, shardingTotalCount)
                    .shardingItemParameters(shardingItemParameters)
                    .description(description)
                    .failover(failover)
                    .jobParameter(jobParameter)
                    .misfire(misfire)
                    .jobProperties(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(), jobExceptionHandler)
                    .jobProperties(JobProperties.JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER.getKey(), executorServiceHandler)
                    .build();

            // 根据实现类类型创建不同类型的JOB
            JobTypeConfiguration typeConfiguration = null;
            if (ElasticJobTypeEnum.SIMPLE.getType().equals(jobTypeName)) {
                typeConfiguration = new SimpleJobConfiguration(coreConfiguration, className);
            } else if (ElasticJobTypeEnum.DATAFLOW.getType().equals(jobTypeName)) {
                typeConfiguration = new DataflowJobConfiguration(coreConfiguration, className, streamingProcess);
            } else if (ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
                typeConfiguration = new ScriptJobConfiguration(coreConfiguration, scriptCommandLine);
            }

            // 创建LiteJobConfiguration
            LiteJobConfiguration jobConfiguration = LiteJobConfiguration
                    .newBuilder(typeConfiguration)
                    .overwrite(overwrite)
                    .disabled(disabled)
                    .monitorPort(monitorPort)
                    .monitorExecution(monitorExecution)
                    .maxTimeDiffSeconds(maxTimeDiffSeconds)
                    .jobShardingStrategyClass(jobShardingStrategyClass)
                    .reconcileIntervalMinutes(reconcileIntervalMinutes)
                    .build();

            // 创建一个Spring的beanDefinition
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);
            factory.setInitMethodName("init");
            factory.setScope("prototype");
            // 添加bean构造参数
            if (!ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
                factory.addConstructorArgValue(bean);
            }
            factory.addConstructorArgValue(zookeeperRegistryCenter);
            factory.addConstructorArgValue(jobConfiguration);
            // 如果有eventTraceRdbDataSource，也需进行添加
            if (StringUtils.hasText(eventTraceRdbDataSource)) {
                BeanDefinitionBuilder rdbFactory = BeanDefinitionBuilder.rootBeanDefinition(JobEventRdbConfiguration.class);
                rdbFactory.addConstructorArgReference(eventTraceRdbDataSource);
                factory.addConstructorArgValue(rdbFactory.getBeanDefinition());
            }

            // 添加监听
            List<?> elasticJobListeners = getTargetElasticJobListeners(elasticJobConfig);
            factory.addConstructorArgValue(elasticJobListeners);

            // 将SpringJobScheduler注入到Spring容器中
            DefaultListableBeanFactory listableBeanFactory = (DefaultListableBeanFactory) appCtx.getAutowireCapableBeanFactory();
            String registerBeanName = elasticJobConfig.jobName() + "SpringJobScheduler";
            listableBeanFactory.registerBeanDefinition(registerBeanName, factory.getBeanDefinition());
            SpringJobScheduler springJobScheduler = (SpringJobScheduler) appCtx.getBean(registerBeanName);
            springJobScheduler.init();
            log.info("elastic job #{}# is started... ", jobName);
        }
        log.info("elastic-job had started {} jobs", beanMap.size());
    }

    private List<BeanDefinition> getTargetElasticJobListeners(ElasticJobConfig conf) {
        List<BeanDefinition> result = new ManagedList<>(2);
        String listeners = conf.listener();
        if (StringUtils.hasText(listeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(listeners);
            factory.setScope("prototype");
            result.add(factory.getBeanDefinition());
        }

        String distributedListeners = conf.distributedListener();
        long startedTimeoutMilliseconds = conf.startedTimeoutMilliseconds();
        long completedTimeoutMilliseconds = conf.completedTimeoutMilliseconds();

        if (StringUtils.hasText(distributedListeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(distributedListeners);
            factory.setScope("prototype");
            factory.addConstructorArgValue(startedTimeoutMilliseconds);
            factory.addConstructorArgValue(completedTimeoutMilliseconds);
            result.add(factory.getBeanDefinition());
        }
        return result;
    }

    private Class<?> getClassByName(String className) {
        Class<?> clazz = null;
        try {
            clazz = Class.forName(className);
        } catch (ClassNotFoundException e) {
            log.error("class not found >> class: {}", className);
            e.printStackTrace();
        }
        return clazz;
    }
}
